package org.qpcr.community.device.service;

import com.alibaba.fastjson.JSONObject;
import io.r2dbc.spi.ConnectionFactory;
import io.swagger.v3.oas.annotations.Parameter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
import org.hswebframework.ezorm.rdb.mapping.ReactiveUpdate;
import org.hswebframework.ezorm.rdb.mapping.defaults.SaveResult;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.authorization.Authentication;
import org.hswebframework.web.crud.events.EntityDeletedEvent;
import org.hswebframework.web.crud.events.EntityEventHelper;
import org.hswebframework.web.crud.query.QueryHelper;
import org.hswebframework.web.crud.service.GenericReactiveCrudService;
import org.hswebframework.web.exception.BusinessException;
import org.hswebframework.web.exception.I18nSupportException;
import org.hswebframework.web.exception.TraceSourceException;
import org.hswebframework.web.i18n.LocaleUtils;
import org.hswebframework.web.id.IDGenerator;
import org.qpcr.community.device.entity.DeviceInstanceEntity;
import org.qpcr.community.device.entity.DeviceProductEntity;
import org.qpcr.community.device.entity.DeviceProperty;
import org.qpcr.community.device.entity.DeviceStateInfo;
import org.qpcr.community.device.entity.DeviceTagEntity;
import org.qpcr.community.device.entity.vo.DeviceInstanceVo;
import org.qpcr.community.device.enums.Client;
import org.qpcr.community.device.enums.DeviceState;
import org.qpcr.community.device.enums.SyncState;
import org.qpcr.community.device.events.DeviceDeployedEvent;
import org.qpcr.community.device.events.DeviceUnregisterEvent;
import org.qpcr.community.device.response.DeviceDeployResult;
import org.qpcr.community.device.response.DeviceDeptRelateResponse;
import org.qpcr.community.device.response.DeviceDetail;
import org.qpcr.community.device.response.ResetDeviceConfigurationResult;
import org.qpcr.community.relation.RelationObjectProvider;
import org.qpcr.community.relation.service.RelationService;
import org.qpcr.community.relation.service.response.RelatedInfo;
import org.qpcr.community.utils.ErrorUtils;
import org.qpcr.core.device.DeviceConfigKey;
import org.qpcr.core.device.DeviceOperator;
import org.qpcr.core.device.DeviceProductOperator;
import org.qpcr.core.device.DeviceRegistry;
import org.qpcr.core.enums.ErrorCode;
import org.qpcr.core.event.EventBus;
import org.qpcr.core.exception.DeviceOperationException;
import org.qpcr.core.message.DeviceMessageReply;
import org.qpcr.core.message.FunctionInvokeMessageSender;
import org.qpcr.core.message.ReadPropertyMessageSender;
import org.qpcr.core.message.WritePropertyMessageSender;
import org.qpcr.core.message.codec.Transport;
import org.qpcr.core.message.function.FunctionInvokeMessageReply;
import org.qpcr.core.message.property.ReadPropertyMessageReply;
import org.qpcr.core.message.property.WritePropertyMessageReply;
import org.qpcr.core.metadata.ConfigMetadata;
import org.qpcr.core.metadata.DeviceMetadata;
import org.qpcr.core.metadata.MergeOption;
import org.qpcr.core.utils.CyclicDependencyChecker;
import org.qpcr.reactor.ql.utils.CastUtils;
import org.qpcr.supports.official.TyLinksDeviceMetadataCodec;
import org.reactivestreams.Publisher;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.reactive.TransactionalOperator;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;
import reactor.util.function.Tuples;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
@Slf4j
public class LocalDeviceInstanceService extends GenericReactiveCrudService<DeviceInstanceEntity, String> {

    private final DeviceRegistry registry;

    private final LocalDeviceProductService deviceProductService;
    @Resource
    private DeviceModuleStateService deviceModuleStateService;

    private final ReactiveRepository<DeviceTagEntity, String> tagRepository;

    private final ApplicationEventPublisher eventPublisher;

    private final DeviceConfigMetadataManager metadataManager;

    private final RelationService relationService;

    private final TransactionalOperator transactionalOperator;

    private final ConnectionFactory connectionFactory;

    private final QueryHelper queryHelper;
    private final EventBus eventBus;

    public LocalDeviceInstanceService(DeviceRegistry registry,
                                      LocalDeviceProductService deviceProductService,
                                      @SuppressWarnings("all")
                                      ReactiveRepository<DeviceTagEntity, String> tagRepository,
                                      ApplicationEventPublisher eventPublisher,
                                      DeviceConfigMetadataManager metadataManager,
                                      RelationService relationService,
                                      TransactionalOperator transactionalOperator,
                                      QueryHelper queryHelper,
                                      EventBus eventBus, ConnectionFactory connectionFactory) {
        this.registry = registry;
        this.deviceProductService = deviceProductService;
        this.tagRepository = tagRepository;
        this.eventPublisher = eventPublisher;
        this.metadataManager = metadataManager;
        this.relationService = relationService;
        this.transactionalOperator = transactionalOperator;
        this.queryHelper = queryHelper;
        this.eventBus = eventBus;
        this.connectionFactory = connectionFactory;
    }

    @Override
    public Mono<SaveResult> save(Publisher<DeviceInstanceEntity> entityPublisher) {
        return Flux
            .from(entityPublisher)
            .flatMap(instance -> {
                instance.setState(null);
                if (StringUtils.isEmpty(instance.getId())) {
                    return handleCreateBefore(instance);
                }
                return registry
                    .getDevice(instance.getId())
                    .flatMap(DeviceOperator::getState)
                    .map(DeviceState::of)
                    .onErrorReturn(DeviceState.offline)
                    .defaultIfEmpty(DeviceState.notActive)
                    .doOnNext(instance::setState)
                    .thenReturn(instance);
            })
            .as(super::save);
    }

    private Flux<DeviceInstanceEntity> findByProductId(String productId) {
        return createQuery()
            .and(DeviceInstanceEntity::getProductId, productId)
            .fetch();
    }
//    private Flux<DeviceInstanceEntity> findByWhere(String productId) {
//        String
//        return createQuery()
//            .leftJoin("device_info t2", "t1.device_id = t2.id")  // 假设 t1 是主表别名
//            .select("t1.*, t2.info_field as extraInfo")          // 选择需要的字段
//            .fetch();
//    }


    private Set<String> getProductConfigurationProperties(DeviceProductEntity product) {
        if (MapUtils.isNotEmpty(product.getConfiguration())) {
            return product.getConfiguration()
                          .keySet();
        }
        return new HashSet<>();
    }

    private Mono<Map<String, Object>> resetConfiguration(DeviceProductEntity product, DeviceInstanceEntity device) {
        return metadataManager
            .getProductConfigMetadataProperties(product.getId())
            .defaultIfEmpty(getProductConfigurationProperties(product))
            .flatMap(set -> {
                if (set.size() > 0) {
                    if (MapUtils.isNotEmpty(device.getConfiguration())) {
                        set.forEach(device.getConfiguration()::remove);
                    }
                    //重置注册中心里的配置
                    return registry
                        .getDevice(device.getId())
                        .flatMap(opts -> opts.removeConfigs(set))
                        .then();
                }
                return Mono.empty();
            })
            .then(
                //更新数据库
                createUpdate()
                    .set(device::getConfiguration)
                    .where(device::getId)
                    .execute()
            )
            .then(Mono.fromSupplier(device::getConfiguration));
    }

    /**
     * 重置设备配置
     *
     * @param deviceId 设备ID
     * @return 重置后的配置
     * @since 1.2
     */
    public Mono<Map<String, Object>> resetConfiguration(String deviceId) {
        return this
            .findById(deviceId)
            .zipWhen(device -> deviceProductService.findById(device.getProductId()))
            .flatMap(tp2 -> {
                DeviceProductEntity product = tp2.getT2();
                DeviceInstanceEntity device = tp2.getT1();
                return resetConfiguration(product, device);
            })
            .defaultIfEmpty(Collections.emptyMap());
    }

    public Mono<Long> resetConfiguration(Flux<String> payload) {
        return payload
            .flatMap(deviceId -> resetConfiguration(deviceId)).count();
    }

    /**
     * 重置设备配置信息(根据产品批量重置，性能欠佳，慎用)
     *
     * @param productId 产品ID
     * @return 数量
     */
    public Flux<ResetDeviceConfigurationResult> resetConfigurationByProductId(String productId) {
        return deviceProductService
            .findById(productId)
            .flatMapMany(product -> this
                .findByProductId(productId)
                .flatMap(device -> {
                    return resetConfiguration(product, device)
                        .thenReturn(ResetDeviceConfigurationResult
                                        .success(SaveResult.of(0, 1)))
                        .onErrorResume(throwable -> {
                            String message = device.getId() + ":" + throwable.getMessage();
                            return Mono.just(ResetDeviceConfigurationResult.error(message));
                        });
                })
            );
    }

    /**
     * 发布设备到设备注册中心
     *
     * @param id 设备ID
     * @return 发布结果
     */
    public Mono<DeviceDeployResult> deploy(String id) {
        return findById(id)
            .flux()
            .as(flux -> deploy(flux, Mono::error))
            .singleOrEmpty();
    }


    /**
     * 批量发布设备到设备注册中心,并异常返回空
     *
     * @param flux 设备实例流
     * @return 发布数量
     */
    public Flux<DeviceDeployResult> deploy(Flux<DeviceInstanceEntity> flux) {
        return this
            .deploy(flux, err -> Mono.empty());
//            .contextWrite(TraceSourceException.deepTraceContext());
    }

    /**
     * 批量发布设备到设备注册中心并指定异常
     *
     * @param flux 设备实例流
     * @return 发布数量
     */
    public Flux<DeviceDeployResult> deploy(Flux<DeviceInstanceEntity> flux, Function<Throwable, Mono<Void>> fallback) {
        //设备回滚 key: deviceId value: 操作
        Map<String, Mono<Void>> rollback = new ConcurrentHashMap<>();

        return flux
            //添加回滚操作,用于再触发DeviceDeployedEvent事件执行失败时进行回滚.
            .flatMap(device -> registry
                .getDevice(device.getId())
                .switchIfEmpty(Mono.fromRunnable(() -> {
                    //设备之前没有注册的回滚操作(注销)
                    rollback.put(device.getId(), registry.unregisterDevice(device.getId()));
                }))
                .thenReturn(device))
            //发布到注册中心
            .flatMap(instance -> registry
                .register(instance.toDeviceInfo())
                .flatMap(deviceOperator -> deviceOperator
                    .checkState()//激活时检查设备状态
                    .onErrorReturn(org.qpcr.core.device.DeviceState.offline)
                    .flatMap(r -> {
                        if (r.equals(org.qpcr.core.device.DeviceState.unknown) ||
                            r.equals(org.qpcr.core.device.DeviceState.noActive)) {
                            instance.setState(DeviceState.offline);
                            return deviceOperator.putState(org.qpcr.core.device.DeviceState.offline);
                        }
                        instance.setState(DeviceState.of(r));
                        return Mono.just(true);
                    })
                    .flatMap(success -> success ? Mono.just(deviceOperator) : Mono.empty()))
                .thenReturn(instance)
                //激活失败,忽略错误,继续处理其他设备
                .onErrorResume(e -> fallback.apply(e).then(Mono.empty()))
            )
            .buffer(200)//每200条数据批量更新
            .publishOn(Schedulers.single())
            .concatMap(all -> Flux
                .fromIterable(all)
                .groupBy(DeviceInstanceEntity::getState)
                .flatMap(group -> group
                    .map(DeviceInstanceEntity::getId)
                    .collectList()
                    .flatMap(list -> createUpdate()
                        .where()
                        .set(DeviceInstanceEntity::getState, group.key())
                        .set(DeviceInstanceEntity::getRegistryTime, System.currentTimeMillis())
                        .in(DeviceInstanceEntity::getId, list)
                        .is(DeviceInstanceEntity::getState, DeviceState.notActive)
                        .execute()
                        .map(r -> DeviceDeployResult.success(list.size()))))
                //推送激活事件
                .flatMap(res -> DeviceDeployedEvent.of(all).publish(eventPublisher).thenReturn(res))
                //传递国际化上下文
                .as(LocaleUtils::transform)
                .as(transactionalOperator::transactional)
                .onErrorResume(err -> Flux
                    .fromIterable(all)
                    .mapNotNull(device -> rollback.get(device.getId()))
                    .flatMap(Function.identity())
                    .then(
                        Mono.zip(
                            I18nSupportException.tryGetLocalizedMessageReactive(err),
                            TraceSourceException.tryGetOperationLocalizedReactive(err).defaultIfEmpty(""),
                            (msg, opt) -> new DeviceDeployResult(all.size(),
                                                                 false,
                                                                 msg,
                                                                 TraceSourceException.tryGetSource(err),
                                                                 opt))
                    )
                    .flatMap(res -> fallback.apply(err).thenReturn(res))
                )
            )
            //激活时不触发事件,单独处理DeviceDeployedEvent
            .as(EntityEventHelper::setDoNotFireEvent)
            ;
    }

    /**
     * 注销设备,取消后,设备无法再连接到服务. 注册中心也无法再获取到该设备信息.
     *
     * @param id 设备ID
     * @return 注销结果
     */
    public Mono<Integer> unregisterDevice(String id) {
        return this
            .unregisterDevice(Mono.just(id))
            .thenReturn(1);
    }

    /**
     * 批量注销设备
     *
     * @param ids 设备ID
     * @return 注销结果
     */
    public Mono<Integer> unregisterDevice(Publisher<String> ids) {
        return Flux
            .from(ids)
            .buffer(200)
            //先修改状态
            .flatMap(list -> this
                .findById(list)
                .collectList()
                .flatMap(devices -> DeviceUnregisterEvent.of(devices).publish(eventPublisher))
                .then(this
                          .createUpdate()
                          .set(DeviceInstanceEntity::getState, DeviceState.notActive.getValue())
                          .where().in(DeviceInstanceEntity::getId, list)
                          .execute()
                          .thenReturn(list)
                ))
            .flatMapIterable(Function.identity())
            //再注销
            .flatMap(id -> registry
                .getDevice(id)
                .flatMap(DeviceOperator::disconnect)
                .onErrorResume(err -> Mono.empty())
                .then(registry.unregisterDevice(id))
                .onErrorResume(err -> Mono.empty())
                .thenReturn(id))
            .count()
            .map(Long::intValue)
            //注销不触发事件,单独处理DeviceDeployedEvent
            .as(EntityEventHelper::setDoNotFireEvent);
    }

    @Override
    public Mono<Integer> deleteById(Publisher<String> idPublisher) {
        return Flux.from(idPublisher)
                   .collectList()
                   .flatMap(list -> createDelete()
                       .where()
                       .in(DeviceInstanceEntity::getId, list)
                       .and(DeviceInstanceEntity::getState, DeviceState.notActive)
                       .execute());
    }

    private boolean hasContext(QueryParamEntity param, String key) {
        return param
            .getContext(key)
            .map(CastUtils::castBoolean)
            .orElse(true);
    }

    //分页查询设备详情列表
    public Mono<PagerResult<DeviceDetail>> queryDeviceDetail(QueryParamEntity entity) {

        return this
            .queryPager(entity)
            .filter(e -> CollectionUtils.isNotEmpty(e.getData()))
            .flatMap(result -> this
                .convertDeviceInstanceToDetail(result.getData(),
                                               hasContext(entity, "includeTags"),
                                               hasContext(entity, "includeBind"),
                                               hasContext(entity, "includeRelations"),
                                               hasContext(entity, "includeFirmwareInfos"))
                .collectList()
                .map(detailList -> PagerResult.of(result.getTotal(), detailList, entity)))
            .defaultIfEmpty(PagerResult.empty());
    }

    //查询设备详情列表
    public Flux<DeviceDetail> queryDeviceDetailList(QueryParamEntity entity) {
        return this
            .query(entity)
            .collectList()
            .flatMapMany(list -> this
                .convertDeviceInstanceToDetail(list,
                                               hasContext(entity, "includeTags"),
                                               hasContext(entity, "includeBind"),
                                               hasContext(entity, "includeRelations"),
                                               hasContext(entity, "includeFirmwareInfos")));
    }

    private Mono<Map<String, List<DeviceTagEntity>>> queryDeviceTagGroup(Collection<String> deviceIdList) {
        return tagRepository
            .createQuery()
            .where()
            .in(DeviceTagEntity::getDeviceId, deviceIdList)
            .fetch()
            .collect(Collectors.groupingBy(DeviceTagEntity::getDeviceId))
            .defaultIfEmpty(Collections.emptyMap());
    }

    private Flux<DeviceDetail> convertDeviceInstanceToDetail(List<DeviceInstanceEntity> instanceList,
                                                             boolean includeTag,
                                                             boolean includeBinds,
                                                             boolean includeRelations,
                                                             boolean includeFirmwareInfos) {
        if (CollectionUtils.isEmpty(instanceList)) {
            return Flux.empty();
        }
        List<String> deviceIdList = new ArrayList<>(instanceList.size());
        //按设备产品分组
        Map<String, List<DeviceInstanceEntity>> productGroup = instanceList
            .stream()
            .peek(device -> deviceIdList.add(device.getId()))
            .collect(Collectors.groupingBy(DeviceInstanceEntity::getProductId));
        //标签
        Mono<Map<String, List<DeviceTagEntity>>> tags = includeTag
            ? this.queryDeviceTagGroup(deviceIdList)
            : Mono.just(Collections.emptyMap());

        //关系信息
        Mono<Map<String, List<RelatedInfo>>> relations = includeRelations ? relationService
            .getRelationInfo(RelationObjectProvider.TYPE_DEVICE, deviceIdList)
            .collect(Collectors.groupingBy(RelatedInfo::getObjectId))
            .defaultIfEmpty(Collections.emptyMap())
            : Mono.just(Collections.emptyMap());


        return Mono
            .zip(
                //T1:查询出所有设备的产品信息
                deviceProductService
                    .findById(productGroup.keySet())
                    .collect(Collectors.toMap(DeviceProductEntity::getId, Function.identity())),
                //T2:查询出标签并按设备ID分组
                tags,
                //T3: 关系信息
                relations
            )
            .flatMapMany(tp5 -> Flux
                //遍历设备,将设备信息转为详情.
                .fromIterable(instanceList)
                .flatMap(instance -> this
                    .createDeviceDetail(
                        // 设备
                        instance
                        //产品
                        , tp5.getT1().get(instance.getProductId())
                        //标签
                        , tp5.getT2().get(instance.getId())
                        //关系信息
                        , tp5.getT3().get(instance.getId())
                    )
                ))
            //createDeviceDetail是异步操作,可能导致顺序错乱.进行重新排序.
            .sort(Comparator.comparingInt(detail -> deviceIdList.indexOf(detail.getId())))
            ;
    }

    private Mono<DeviceDetail> createDeviceDetail(DeviceInstanceEntity device,
                                                  DeviceProductEntity product,
                                                  List<DeviceTagEntity> tags,
                                                  List<RelatedInfo> relations) {
        if (product == null) {
            log.warn("device [{}] product [{}] does not exists", device.getId(), device.getProductId());
            return Mono.empty();
        }
        DeviceDetail detail = new DeviceDetail()
            .with(product)
            .with(device)
            .with(tags)
            .withRelation(relations);

        return Mono
            .zip(
                //产品注册信息
                registry
                    .getProduct(product.getId()),
                //feature信息
                metadataManager
                    .getProductFeatures(product.getId())
                    .collectList())
            .flatMap(t2 -> {
                //填充产品中feature信息
                detail.withFeatures(t2.getT2());
                //填充注册中心里的产品信息
                return detail.with(t2.getT1());
            })
            .then(Mono.zip(
                //设备信息
                registry
                    .getDevice(device.getId())
                    //先刷新配置缓存
                    .flatMap(operator -> operator.refreshAllConfig().thenReturn(operator))
                    .flatMap(operator -> operator
                        //检查设备的真实状态,可能出现设备已经离线,但是数据库状态未及时更新的.
                        .checkState()
                        .map(DeviceState::of)
                        //检查失败,则返回原始状态
                        .onErrorReturn(device.getState())
                        //如果状态不一致,则需要更新数据库中的状态
                        .filter(state -> state != detail.getState())
                        .doOnNext(detail::setState)
                        .flatMap(state -> createUpdate()
                            .set(DeviceInstanceEntity::getState, state)
                            .where(DeviceInstanceEntity::getId, device.getId())
                            .execute())
                        .thenReturn(operator)),
                //配置定义
                metadataManager
                    .getDeviceConfigMetadata(device.getId())
                    .flatMapIterable(ConfigMetadata::getProperties)
                    .collectList()
            ))
            //填充详情信息
            .flatMap(tp2 -> detail
                .with(tp2.getT1(), tp2.getT2()))
            .switchIfEmpty(
                Mono.defer(() -> {
                    //如果设备注册中心里没有设备信息,并且数据库里的状态不是未激活.
                    //可能是因为注册中心信息丢失,修改数据库中的状态信息.
                    if (detail.getState() != DeviceState.notActive) {
                        return createUpdate()
                            .set(DeviceInstanceEntity::getState, DeviceState.notActive)
                            .where(DeviceInstanceEntity::getId, detail.getId())
                            .execute()
                            .thenReturn(detail.notActive());
                    }
                    return Mono.just(detail.notActive());
                }).thenReturn(detail))
            .onErrorResume(err -> {
                log.warn("get device detail error", err);
                return Mono.just(detail);
            });
    }

    public Mono<DeviceDetail> getDeviceDetail(String deviceId) {

        return this
            .findById(deviceId)
            .map(Collections::singletonList)
            .flatMapMany(list -> convertDeviceInstanceToDetail(list, true, true, true, true))
            .next();
    }

    public Mono<DeviceState> getDeviceState(String deviceId) {
        return registry.getDevice(deviceId)
                       .flatMap(DeviceOperator::checkState)
                       .flatMap(state -> {
                           DeviceState deviceState = DeviceState.of(state);
                           return createUpdate()
                               .set(DeviceInstanceEntity::getState, deviceState)
                               .where(DeviceInstanceEntity::getId, deviceId)
                               .execute()
                               .thenReturn(deviceState);
                       })
                       .defaultIfEmpty(DeviceState.notActive);
    }

    //获取设备属性
    @SneakyThrows
    public Mono<Map<String, Object>> readProperty(String deviceId,
                                                  String property) {
        return registry
            .getDevice(deviceId)
            .switchIfEmpty(ErrorUtils.notFound("error.device_not_found_or_not_activated"))
            .map(DeviceOperator::messageSender)//发送消息到设备
            .map(sender -> sender.readProperty(property).messageId(IDGenerator.SNOW_FLAKE_STRING.generate()))
            .flatMapMany(ReadPropertyMessageSender::send)
            .flatMap(mapReply(ReadPropertyMessageReply::getProperties))
            .reduceWith(LinkedHashMap::new, (main, map) -> {
                main.putAll(map);
                return main;
            });

    }

    //获取标准设备属性
    @SneakyThrows
    public Mono<DeviceProperty> readAndConvertProperty(String deviceId,
                                                       String property) {
        return registry
            .getDevice(deviceId)
            .switchIfEmpty(ErrorUtils.notFound("error.device_not_found_or_not_activated"))
            .flatMap(deviceOperator -> deviceOperator
                .messageSender()
                .readProperty(property)
                .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                .send()
                .flatMap(mapReply(ReadPropertyMessageReply::getProperties))
                .reduceWith(LinkedHashMap::new, (main, map) -> {
                    main.putAll(map);
                    return main;
                })
                .flatMap(map -> {
                    Object value = map.get(property);
                    return deviceOperator
                        .getMetadata()
                        .map(deviceMetadata -> DeviceProperty.of(value, deviceMetadata.getPropertyOrNull(property)));
                }));

    }

    //设置设备属性
    @SneakyThrows
    public Mono<Map<String, Object>> writeProperties(String deviceId,
                                                     Map<String, Object> properties) {

        return registry
            .getDevice(deviceId)
            .switchIfEmpty(ErrorUtils.notFound("error.device_not_found_or_not_activated"))
            .flatMap(operator -> operator
                .messageSender()
                .writeProperty()
                .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                .write(properties)
                .validate()
            )
            .flatMapMany(WritePropertyMessageSender::send)
            .flatMap(mapReply(WritePropertyMessageReply::getProperties))
            .reduceWith(LinkedHashMap::new, (main, map) -> {
                main.putAll(map);
                return main;
            });
    }

    //设备功能调用
    @SneakyThrows
    public Flux<?> invokeFunction(String deviceId,
                                  String functionId,
                                  Map<String, Object> properties) {
        return invokeFunction(deviceId, functionId, properties, true);
    }

    //设备功能调用
    @SneakyThrows
    public Flux<?> invokeFunction(String deviceId,
                                  String functionId,
                                  Map<String, Object> properties,
                                  boolean convertReply) {
        return registry
            .getDevice(deviceId)
            .switchIfEmpty(ErrorUtils.notFound("error.device_not_found_or_not_activated"))
            .flatMap(operator -> operator
                .messageSender()
                .invokeFunction(functionId)
                .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                .setParameter(properties)
                .validate()
            )
            .flatMapMany(FunctionInvokeMessageSender::send)
            .flatMap(convertReply ? mapReply(FunctionInvokeMessageReply::getOutput) : Mono::just);


    }

    //获取设备所有属性
    @SneakyThrows
    public Mono<Map<String, Object>> readProperties(String deviceId, List<String> properties) {

        return registry.getDevice(deviceId)
                       .switchIfEmpty(ErrorUtils.notFound("error.device_not_found_or_not_activated"))
                       .map(DeviceOperator::messageSender)
                       .flatMapMany((sender) -> sender.readProperty()
                                                      .read(properties)
                                                      .messageId(IDGenerator.SNOW_FLAKE_STRING.generate())
                                                      .send())
                       .flatMap(mapReply(ReadPropertyMessageReply::getProperties))
                       .reduceWith(LinkedHashMap::new, (main, map) -> {
                           main.putAll(map);
                           return main;
                       });
    }

    private static <R extends DeviceMessageReply, T> Function<R, Mono<T>> mapReply(Function<R, T> function) {
        return reply -> {
            if (ErrorCode.REQUEST_HANDLING.name().equals(reply.getCode())) {
                throw new DeviceOperationException(ErrorCode.REQUEST_HANDLING, reply.getMessage());
            }
            if (!reply.isSuccess()) {
                if (StringUtils.isEmpty(reply.getMessage())) {
                    throw new BusinessException("error.reply_is_error");
                }
                throw new BusinessException(reply.getMessage(), reply.getCode());
            }
            return Mono.justOrEmpty(function.apply(reply));
        };
    }

    /**
     * 批量同步设备状态
     *
     * @param batch 设备状态ID流
     * @param force 是否强制获取设备状态,强制获取会去设备连接到服务器检查设备是否真实在线
     * @return 同步数量
     */
    public Flux<List<DeviceStateInfo>> syncStateBatch(Flux<List<String>> batch, boolean force) {

        return batch
            .concatMap(list -> Flux
                .fromIterable(list)
                .publishOn(Schedulers.parallel())
                .flatMap(id -> registry
                    .getDevice(id)
                    .flatMap(operator -> {
                        Mono<Byte> state = force
                            ? operator
                            .checkState()
                            .onErrorResume(err -> operator.getState())
                            : operator.getState();
                        return Mono
                            .zip(
                                state.defaultIfEmpty(org.qpcr.core.device.DeviceState.offline),//状态
                                Mono.just(operator.getDeviceId()), //设备id
                                operator.getConfig(DeviceConfigKey.isGatewayDevice).defaultIfEmpty(false)//是否为网关设备
                            );
                    })
                    //注册中心里不存在设备就认为是未激活.
                    .defaultIfEmpty(Tuples.of(org.qpcr.core.device.DeviceState.noActive, id, false)))
                .collect(Collectors.groupingBy(Tuple2::getT1))
                .flatMapIterable(Map::entrySet)
                .flatMap(group -> {
                    List<String> deviceIdList = group
                        .getValue()
                        .stream()
                        .map(Tuple3::getT2)
                        .collect(Collectors.toList());
                    DeviceState state = DeviceState.of(group.getKey());
                    return
                        //批量修改设备状态
                        getRepository()
                            .createUpdate()
                            .set(DeviceInstanceEntity::getState, state)
                            .where()
                            .in(DeviceInstanceEntity::getId, deviceIdList)
                            .when(state != DeviceState.notActive, where -> where.not(DeviceInstanceEntity::getState, DeviceState.notActive))
                            .execute()
                            .thenReturn(group.getValue().size())
                            .then(Mono.just(
                                deviceIdList
                                    .stream()
                                    .map(id -> DeviceStateInfo.of(id, state))
                                    .collect(Collectors.toList())
                            ));
                }))
            //更新状态不触发事件
            .as(EntityEventHelper::setDoNotFireEvent);
    }


    public Mono<Void> mergeMetadata(String deviceId, DeviceMetadata metadata, MergeOption... options) {

        return Mono
            .zip(this.findById(deviceId)
                     .flatMap(device -> {
                         if (StringUtils.hasText(device.getDeriveMetadata())) {
                             return Mono.just(device.getDeriveMetadata());
                         } else {
                             return deviceProductService
                                 .findById(device.getProductId())
                                 .map(DeviceProductEntity::getMetadata);
                         }
                     })
                     .flatMap(TyLinksDeviceMetadataCodec.getInstance()::decode),
                 Mono.just(metadata),
                 (older, newer) -> older.merge(newer, options)
            )
            .flatMap(TyLinksDeviceMetadataCodec.getInstance()::encode)
            .flatMap(newMetadata -> createUpdate()
                .set(DeviceInstanceEntity::getDeriveMetadata, newMetadata)
                .where(DeviceInstanceEntity::getId, deviceId)
                .execute()
                .then(
                    registry
                        .getDevice(deviceId)
                        .flatMap(device -> device.updateMetadata(newMetadata))
                ))
            .then();
    }

    public Flux<DeviceTagEntity> queryDeviceTag(String deviceId, String... tags) {
        return tagRepository
            .createQuery()
            .where(DeviceTagEntity::getDeviceId, deviceId)
            .when(tags.length > 0, q -> q.in(DeviceTagEntity::getKey, Arrays.asList(tags)))
            .fetch();
    }

    //删除设备时，删除设备标签
    @EventListener
    public void handleDeviceDelete(EntityDeletedEvent<DeviceInstanceEntity> event) {
        event.async(
            Flux.concat(
                Flux
                    .fromIterable(event.getEntity())
                    .flatMap(device -> registry
                        .unregisterDevice(device.getId())
                        .onErrorResume(err -> Mono.empty())
                    )
                    .then(),
                tagRepository
                    .createDelete()
                    .where()
                    .in(DeviceTagEntity::getDeviceId, event
                        .getEntity()
                        .stream()
                        .map(DeviceInstanceEntity::getId)
                        .collect(Collectors.toSet()))
                    .execute()
            )
        );
    }

    @Override
    public Mono<Integer> insert(DeviceInstanceEntity data) {
        return this
            .handleCreateBefore(data)
            .flatMap(super::insert);
    }

    @Override
    public Mono<Integer> insert(Publisher<DeviceInstanceEntity> entityPublisher) {
        return super.insert(Flux.from(entityPublisher).flatMap(this::handleCreateBefore));
    }

    @Override
    public Mono<Integer> insertBatch(Publisher<? extends Collection<DeviceInstanceEntity>> entityPublisher) {
        return Flux.from(entityPublisher)
                   .flatMapIterable(Function.identity())
                   .as(this::insert);
    }

    private Mono<DeviceInstanceEntity> handleCreateBefore(DeviceInstanceEntity instanceEntity) {
        return Mono
            .zip(
                deviceProductService.findById(instanceEntity.getProductId()),
                registry
                    .getProduct(instanceEntity.getProductId())
                    .flatMap(DeviceProductOperator::getProtocol),
                (product, protocol) -> protocol.doBeforeDeviceCreate(
                    Transport.of(product.getTransportProtocol()), instanceEntity.toDeviceInfo(false))
            )
            .flatMap(Function.identity())
            .doOnNext(info -> {
                if (ObjectUtils.isEmpty(instanceEntity.getId())) {
                    instanceEntity.setId(info.getId());
                }
                instanceEntity.mergeConfiguration(info.getConfiguration());
            })
            .thenReturn(instanceEntity);

    }

    private final CyclicDependencyChecker<DeviceInstanceEntity, Void> checker = CyclicDependencyChecker
        .of(DeviceInstanceEntity::getId, DeviceInstanceEntity::getParentId, this::findById);

    public Mono<Void> checkCyclicDependency(DeviceInstanceEntity device) {
        return checker.check(device);
    }

    public Mono<Void> checkCyclicDependency(String id, String parentId) {
        DeviceInstanceEntity instance = new DeviceInstanceEntity();
        instance.setId(id);
        instance.setParentId(parentId);
        return checker.check(instance);
    }

    public Mono<Void> mergeConfiguration(String deviceId,
                                         Map<String, Object> configuration,
                                         Function<ReactiveUpdate<DeviceInstanceEntity>,
                                             ReactiveUpdate<DeviceInstanceEntity>> updateOperation) {
        if (MapUtils.isEmpty(configuration)) {
            return Mono.empty();
        }
        return this
            .findById(deviceId)
            .flatMap(device -> {
                //合并更新配置
                device.mergeConfiguration(configuration);
                return createUpdate()
                    .set(device::getConfiguration)
                    .set(device::getFeatures)
                    .set(device::getDeriveMetadata)
                    .as(updateOperation)
                    .where(device::getId)
                    .execute();
            })
            .then(
                //更新缓存里到信息
                registry
                    .getDevice(deviceId)
                    .flatMap(device -> device.setConfigs(configuration))
            )
            .then();

    }

    /**
     * 更新检测项同步状态
     *
     * @param productId 产品id
     * @param syncState 同步状态
     * @return Mono<Integer>
     */
    public Mono<Integer> updateItemSyncStatusByProductId(List<String> productId, SyncState syncState) {
        return this.createUpdate()
                   .set(DeviceInstanceEntity::getItemSyncState, syncState)
                   .in(DeviceInstanceEntity::getProductId, productId)
                   .execute();
    }

    /**
     * 更新检测序列同步状态
     *
     * @param productId 产品id
     * @param syncState 同步状态
     * @return Mono<Integer>
     */
    public Mono<Integer> updateSeqSyncStatusByProductId(List<String> productId, SyncState syncState) {
        return this.createUpdate()
                   .set(DeviceInstanceEntity::getSeqSyncState, syncState)
                   .in(DeviceInstanceEntity::getProductId, productId)
                   .execute();
    }

    /**
     * 查询未同步的设备信息
     *
     * @return
     */
    public Mono<List<DeviceInstanceEntity>> queryNotSyncList() {
        return this.createQuery()
                   .or("item_sync_state", SyncState.notSync)
                   .or("seq_sync_state", SyncState.notSync)
                   .or("item_sync_state", null)
                   .or("seq_sync_state", null)
                   .fetch()
                   .map(Function.identity())
                   .collectList();
    }

    /**
     * 查询未同步的设备的检测项和序列信息
     *
     * @return
     */
    public Mono<Integer> updateStateHasSync(List<String> deviceIdList) {
        return this.findById(deviceIdList).collectList()
                   .flatMap(list -> {
                       List<String> productIdList = list
                           .stream()
                           .map(DeviceInstanceEntity::getProductId)
                           .collect(Collectors.toList());
                       this.updateItemSyncStatusByProductId(productIdList, SyncState.hasSync);
                       this.updateSeqSyncStatusByProductId(productIdList, SyncState.hasSync);
                       return Mono.just(0);
                   });
    }


    /**
     * 根据部门ID查询设备是否可用【平板端查询设备列表接口】
     * .flatMap(
     *
     * @return
     */
    public Mono<PagerResult<DeviceInstanceEntity>> queryDeviceByDept(String deptId, Integer type, Integer index, Integer pageSize) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = "  select * from dev_device_instance dev where dev.id in( select device_id from biz_relate_dept_device rea where rea.dept_id ='" + deptId + "')  ";
                if (type != null) {
                    if (type == 1) {//不可连接的设备
                        sql += " and  ((dev.type=1 and dev.account!='" + auth
                            .getUser()
                            .getUsername() + "')   or  dev.state='offline') ";
                    } else if (type == 0) {//可连接的设备 ，包含自己连接的设备
                        sql += " and (dev.type is null or dev.type=0 or (dev.type=1 and dev.account='" + auth
                            .getUser()
                            .getUsername() + "')) and  dev.state='online' ";
                    }
                }
                return queryHelper
                    .select(sql,
                            DeviceInstanceEntity::new)
                    .fetchPaged(index == null ? 0 : index, pageSize == null ? 10 : pageSize);
            });


    }

    /**
     * 根据当前登入人信息查询设备是否可用【平板端查询设备列表接口】
     * .flatMap(
     *
     * @return
     */
    public Mono<PagerResult<DeviceInstanceEntity>> queryDeviceByUser(Integer type, Integer index, Integer pageSize) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = "  select * from dev_device_instance dev where dev.id in( select device_id from biz_relate_dept_device rea where rea.dept_id in (select d.dimension_id from s_dimension_user d where d.dimension_type_id='org' and user_id='" + auth
                    .getUser()
                    .getId() + "') )  ";
                if (type != null) {
                    if (type == 1) {//不可连接的设备
                        sql += " and  ((dev.type=1 and dev.account!='" + auth.getUser().getUsername() + "')  " +
                            "or ( dev.type=1 and dev.account= '" + auth
                            .getUser()
                            .getUsername() + "' and dev.last_login_client!='" + Client.APP.getClientId() + "')    " +
                            "or  dev.state='offline' ) ";
                    } else if (type == 0) {//可连接的设备 ，包含自己连接的设备
                        sql += " and (dev.type is null or dev.type=0 or (dev.type=1 and dev.account='" + auth
                            .getUser()
                            .getUsername() + "' and dev.last_login_client='" + Client.APP.getClientId() + "' )) and  dev.state='online' ";
                    }
                }
                return queryHelper
                    .select(sql,
                            DeviceInstanceEntity::new)
                    .fetchPaged(index == null ? 0 : index, pageSize == null ? 10 : pageSize);
            });


    }

    /**
     * 根据当前登入人信息查询已连接设备数
     * .flatMap(
     *
     * @return
     */
    public Mono<Integer> connectDeviceCount() {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = "  select dev.id from dev_device_instance dev where dev.id in( select device_id from biz_relate_dept_device rea where rea.dept_id in (select d.dimension_id from s_dimension_user d where d.dimension_type_id='org' and user_id='" + auth
                    .getUser()
                    .getId() + "') )  ";
                sql += " and (dev.type=1 and dev.account='" + auth
                    .getUser()
                    .getUsername() + "' and dev.last_login_client='" + Client.APP.getClientId() + "' ) and  dev.state='online' ";
                return queryHelper
                    .select(sql, String::new).count();
            });
    }


    /**
     * 根据部门ID查询设备是否可用【管理端绑定设备接口】
     * .flatMap(
     *
     * @return
     */
    public Mono<PagerResult<DeviceInstanceEntity>> queryDeviceNoRelate(Integer index, Integer pageSize) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = "   select * from dev_device_instance dev where dev.id not in( select device_id from biz_relate_dept_device rea )  ";

                return queryHelper
                    .select(sql,
                            DeviceInstanceEntity::new)
                    .fetchPaged(index == null ? 0 : index, pageSize == null ? 10 : pageSize);
            });


    }

    /**
     * 根据部门ID查询已绑定的设备
     * .flatMap(
     *
     * @return
     */
    public Mono<PagerResult<DeviceDeptRelateResponse>> queryDeviceRelate(String deptId, Integer index, Integer pageSize) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = " select rea.*,dev.name,dev.product_name,dev.product_id  from biz_relate_dept_device rea  INNER JOIN dev_device_instance dev on dev.id =rea.device_id  where rea.dept_id ='" + deptId + "' ";

                return queryHelper
                    .select(sql,
                            DeviceDeptRelateResponse::new)
                    .fetchPaged(index == null ? 0 : index, pageSize == null ? 10 : pageSize);
            });
    }

    public Mono<PagerResult<DeviceInstanceVo>> queryPagerWithProduct(QueryParamEntity query) {
        return super.queryPager(query)
                    .flatMap(result -> {
                        if (CollectionUtils.isEmpty(result.getData())) {
                            return Mono.just(PagerResult.of(result.getTotal(), Collections.emptyList(), query));
                        }

                        // 提取所有设备的产品ID
                        List<String> productIds = result.getData().stream()
                                                        .map(DeviceInstanceEntity::getProductId)
                                                        .distinct()
                                                        .collect(Collectors.toList());

                        // 查询产品信息
                        return deviceProductService.findById(productIds)
                                                   .collect(Collectors.toMap(DeviceProductEntity::getId, Function.identity()))
                                                   .flatMap(productMap -> {
                                                       // 转换设备实例为VO，并设置classifiedName
                                                       return Flux.fromIterable(result.getData())
                                                                 .flatMap(device -> {
                                                                     DeviceInstanceVo vo = new DeviceInstanceVo();
                                                                     // 复制设备实例的所有属性到VO
                                                                     org.hswebframework.web.bean.FastBeanCopier.copy(device, vo);

                                                                     // 设置classifiedName
                                                                     DeviceProductEntity product = productMap.get(device.getProductId());
                                                                     if (product != null) {
                                                                         vo.setClassifiedName(product.getClassifiedName());
                                                                     }


                                                                     if (vo.getState() == DeviceState.online) {
                                                                         // 使用异步方式获取模块状态
                                                                         return deviceModuleStateService.getModuleStateMono(device.getId())
                                                                                 .map(moduleState -> {
                                                                                     if (null != moduleState) {
                                                                                         switch (moduleState) {
                                                                                             case 1:
                                                                                                 vo.setState(DeviceState.online);
                                                                                                 break;
                                                                                             case 2:
                                                                                                 vo.setState(DeviceState.running);
                                                                                                 break;
                                                                                             case 3:
                                                                                                 vo.setState(DeviceState.online);
                                                                                                 break;
                                                                                             case 4:
                                                                                                 vo.setState(DeviceState.removed);
                                                                                                 break;
                                                                                             default:
                                                                                                 vo.setState(DeviceState.offline);
                                                                                                 break;
                                                                                         }
                                                                                     } else {
                                                                                     }
                                                                                     return vo;
                                                                                 });
                                                                     } else {
                                                                         return Mono.just(vo);
                                                                     }
                                                                 })
                                                                 .collectList()
                                                                 .map(voList -> PagerResult.of(result.getTotal(), voList, query));
                                                   });
                    });
    }

    /**
     * 根据部门ID查询已绑定的设备
     * .flatMap(
     *
     * @return
     */
    public Mono<PagerResult<DeviceInstanceVo>> queryDeviceByWhere(String id, String deptId, String productId, Integer index, Integer pageSize, String state, String name) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                String sql = " select dev_device_instance.`id` as `id` , dev_device_instance.`photo_url` as `photoUrl` , dev_device_instance.`name` as `name` , dev_device_instance.`device_type` as `deviceType` , dev_device_instance.`describe` as `describe` , dev_device_instance.`product_id` as `productId` , dev_device_instance.`product_name` as `productName` , dev_device_instance.`configuration` as `configuration` , dev_device_instance.`derive_metadata` as `deriveMetadata` , dev_device_instance.`state` as `state` , dev_device_instance.`type` as `type` , dev_device_instance.`account` as `account` , dev_device_instance.`log_type` as `logType` , dev_device_instance.`creator_id` as `creatorId` , dev_device_instance.`creator_name` as `creatorName` , dev_device_instance.`create_time` as `createTime` , dev_device_instance.`registry_time` as `registryTime` , dev_device_instance.`org_id` as `orgId` , dev_device_instance.`parent_id` as `parentId` , dev_device_instance.`features` as `features` , dev_device_instance.`modify_time` as `modifyTime` , dev_device_instance.`modifier_id` as `modifierId` , dev_device_instance.`modifier_name` as `modifierName` , dev_device_instance.`item_sync_state` as `itemSyncState` , dev_device_instance.`seq_sync_state` as `seqSyncState` from `dev_device_instance` dev_device_instance   " +

                    " left JOIN  biz_relate_dept_device rea  on dev_device_instance.id =rea.device_id   where 1=1 ";

                if (name != null) {
                    sql += " and dev_device_instance.id like '%" + name + "%'";
                }
                if (id != null) {
                    sql += " and dev_device_instance.id='" + id + "'";
                }
                if (deptId != null) {
                    sql += " and rea.dept_id ='" + deptId + "'";
                }
                if (productId != null) {
                    sql += " and dev_device_instance.product_id='" + productId + "'";
                }
                if (state != null) {
                    sql += " and dev_device_instance.state='" + state + "'";
                }
                return queryHelper
                    .select(sql,
                            DeviceInstanceEntity::new)
                    .fetchPaged(index == null ? 0 : index, pageSize == null ? 10 : pageSize)
                    .flatMap(result -> {
                        QueryParamEntity queryParamEntity = new QueryParamEntity();
                        queryParamEntity.setTotal(result.getTotal());
                        queryParamEntity.setPageIndex(result.getPageIndex());
                        queryParamEntity.setPageSize(result.getPageSize());
                        if (CollectionUtils.isEmpty(result.getData())) {
                            return Mono.just(PagerResult.of(result.getTotal(), Collections.emptyList(), queryParamEntity));
                        }

                        // 提取所有设备的产品ID
                        List<String> productIds = result.getData().stream()
                                                        .map(DeviceInstanceEntity::getProductId)
                                                        .distinct()
                                                        .collect(Collectors.toList());

                        // 查询产品信息
                        return deviceProductService.findById(productIds)
                                                   .collect(Collectors.toMap(DeviceProductEntity::getId, Function.identity()))
                                                   .flatMap(productMap -> {
                                                       // 转换设备实例为VO，并设置classifiedName
                                                       List<DeviceInstanceVo> voList = result.getData().stream()
                                                                                             .map(device -> {
                                                                                                 DeviceInstanceVo vo = new DeviceInstanceVo();
                                                                                                 // 复制设备实例的所有属性到VO
                                                                                                 org.hswebframework.web.bean.FastBeanCopier.copy(device, vo);

                                                                                                 // 设置classifiedName
                                                                                                 DeviceProductEntity product = productMap.get(device.getProductId());
                                                                                                 if (product != null) {
                                                                                                     vo.setClassifiedName(product.getClassifiedName());
                                                                                                 }

                                                                                                 return vo;
                                                                                             })
                                                                                             .collect(Collectors.toList());

                                                       return Mono.just(PagerResult.of(result.getTotal(), voList, queryParamEntity));
                                                   });
                    });
            });
    }

    /**
     * 网络更新设备的账号绑定和断开状态
     *
     * @return
     */
    public Mono<Integer> updateDeviceState(String id, Integer type, Integer logType) {
        return Authentication
            .currentReactive().flatMap(auth -> {
                if (type == 0) {//解除绑定
                    return this.createQuery().where(DeviceInstanceEntity::getId, id).fetchOne().flatMap(instance -> {
                        String account = auth.getUser().getUsername();
                        if (!account.equals(instance.getAccount())) {//如果是解除绑定，需要判断账号是当前的登入账号
                            return Mono.error(new BusinessException("该设备不是本账号登入，不能进行断开连接操作，请确认后再试", -200));
                        } else {
                            return this.createUpdate()
                                       .set(DeviceInstanceEntity::getType, 0)
                                       .setNull(DeviceInstanceEntity::getAccount)
                                       .setNull(DeviceInstanceEntity::getLogType)
                                       .set(DeviceInstanceEntity::getAppLinkState, 0)
                                       .in(DeviceInstanceEntity::getId, id)
                                       .execute()
                                       .flatMap(integer -> {
                                           JSONObject jsonObject = new JSONObject();
                                           jsonObject.put("type", "loginOff");
                                           jsonObject.put("data", account);
                                           JSONObject jsonObject_final = new JSONObject();
                                           jsonObject_final.put("data", jsonObject);
                                           return eventBus
                                               .publish("/device/" + instance.getProductId() + "/" + instance.getId() + "/message/event/eventAll", jsonObject_final)
                                               .thenReturn(integer);
                                       });
                        }
                    });
                } else {//建立连接
                    return this
                        .createQuery()
                        .where(DeviceInstanceEntity::getAccount, auth.getUser().getUsername())
                        .count()
                        .flatMap(count -> {
                            if (count >= 5) {
                                return Mono.error(new BusinessException("一台设备最多只能同时登录5台设备", -200));
                            } else {
                                return this
                                    .createQuery()
                                    .where(DeviceInstanceEntity::getId, id)
                                    .fetchOne()
                                    .flatMap(instance -> {
                                        String account = auth.getUser().getUsername();
                                        if (instance.getType() == 1 && !account.equals(instance.getAccount())) {//建立连接，先判断当前设备是否有被连接
                                            return Mono.error(new BusinessException("该设备已经被其他用户连接绑定，请确认后再试", -200));
                                        } else {
                                            return this.createUpdate()
                                                       .set(DeviceInstanceEntity::getType, 1)
                                                       .set(DeviceInstanceEntity::getAccount, account)
                                                       .set(DeviceInstanceEntity::getLogType, logType)
                                                       .set(DeviceInstanceEntity::getAppLinkState, 1)
                                                       .set(DeviceInstanceEntity::getLastLoginClient, Client.APP.getClientId())
                                                       .in(DeviceInstanceEntity::getId, id)
                                                       .execute()
                                                       .flatMap(integer -> {
                                                           JSONObject jsonObject = new JSONObject();
                                                           jsonObject.put("type", "loginIn");
                                                           jsonObject.put("data", account);
                                                           JSONObject jsonObject_final = new JSONObject();
                                                           jsonObject_final.put("data", jsonObject);
                                                           return eventBus
                                                               .publish("/device/" + instance.getProductId() + "/" + instance.getId() + "/message/event/eventAll", jsonObject_final)
                                                               .thenReturn(integer);
                                                       });
                                        }
                                    });

                            }

                        });
                }

            });
    }

    /**
     * 设备端的账号绑定和断开状态
     *
     * @return
     */
    public Mono<Integer> deviceUpdateState(String id, Integer type, String account) {

        if (type == 0) {//解除绑定
            return this.createQuery().where(DeviceInstanceEntity::getId, id).fetchOne().flatMap(instance -> {
                if (!Client.APP.getClientId().equals(instance.getLastLoginClient())) {
                    return this.createUpdate()
                               .set(DeviceInstanceEntity::getType, 0)
                               .setNull(DeviceInstanceEntity::getAccount)
                               .setNull(DeviceInstanceEntity::getLogType)
                               .in(DeviceInstanceEntity::getId, id)
                               .execute();
                } else {
                    return Mono.error(new BusinessException("该设备已在APP登入，不能进行断开连接操作，请确认后再试", -200));
                }

//                        if(!account.equals(instance.getAccount())){//如果是解除绑定，需要判断账号是当前的登入账号
//                            return  Mono.error(new BusinessException("该设备不是本账号登入，不能进行断开连接操作，请确认后再试", -200));
//                        }else{
//                            return this.createUpdate()
//                                       .set(DeviceInstanceEntity::getType, 0)
//                                       .setNull(DeviceInstanceEntity::getAccount)
//                                       .setNull(DeviceInstanceEntity::getLogType)
//                                       .in(DeviceInstanceEntity::getId, id)
//                                       .execute();
//                        }
            });
        } else {//建立连接
            return this.createQuery().where(DeviceInstanceEntity::getAccount, account).count().flatMap(count -> {
                if (count >= 5) {
                    return Mono.error(new BusinessException("一台设备最多只能同时登录5台设备", -200));
                } else {
                    return this.createQuery().where(DeviceInstanceEntity::getId, id).fetchOne().flatMap(instance -> {

                        if (instance.getType() == 1 && !account.equals(instance.getAccount())) {//建立连接，先判断当前设备是否有被连接
                            return Mono.error(new BusinessException("该设备已经被其他用户连接绑定，请确认后再试", -200));
                        } else {
                            return this.createUpdate()
                                       .set(DeviceInstanceEntity::getType, 1)
                                       .set(DeviceInstanceEntity::getAccount, account)
                                       .set(DeviceInstanceEntity::getLogType, 1)
                                       .set(DeviceInstanceEntity::getLastLoginClient, Client.PCR_SYSTEM.getClientId())
                                       .in(DeviceInstanceEntity::getId, id)
                                       .execute();
                        }
                    });

                }

            });
        }


    }


    /**
     * 更新设备的账号绑定和断开状态
     */
    public Mono<Void> DelDeviceLogState(String id) {
        String sql = "UPDATE dev_device_instance SET account = ?, log_type = ?,type=? WHERE id = ?";

        return Mono.from(connectionFactory.create())
                   .flatMap(connection ->
                                Mono.from(connection.beginTransaction())  // 开启事务
                                    .then(Mono.from(connection.createStatement(sql)
                                                              .bindNull(0, String.class)
                                                              .bindNull(1, Integer.class)
                                                              .bind(2, 0)
                                                              .bind(3, id)
                                                              .execute()))
                                    .flatMap(result -> Mono.from(result.getRowsUpdated()))
                                    .doOnNext(rowsUpdated -> {
                                        if (rowsUpdated == 0) {
                                            System.out.println("No rows updated for id: " + id);
                                        } else {
                                            System.out.println("Successfully updated " + rowsUpdated + " row(s) for id: " + id);
                                        }
                                    })
                                    .then(Mono.from(connection.commitTransaction())) // 提交事务
                                    .doFinally(signalType -> connection.close())     // 关闭连接
                   )
                   .then();

    }


    public Mono<DeviceInstanceEntity> queryDeviceByUserName() {
        return Authentication
            .currentReactive().flatMap(auth -> {
                return this.createQuery()
                           .where(DeviceInstanceEntity::getAccount, auth.getUser().getUsername())
                           .fetchOne();
            });
    }

    public Mono<DeviceInstanceEntity> queryDeviceById(String id) {

        return this.createQuery()
                   .where(DeviceInstanceEntity::getId, id)
                   .fetchOne();

    }


}
